package com.atguigu.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.utils.ClickHouseUtil;
import com.atguigu.gmall.realtime.utils.DateTimeUtil;
import com.atguigu.gmall.realtime.utils.MyKafkaUtil;
import com.atguigu.gmall.realtime.bean.VisitorStats;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;


import java.time.Duration;
import java.util.Date;

/**
 * @Description: dws层访客维度计算
 * @Author: tiancy
 * @Create: 2021/12/5
 */

/*
    TODO dws层访客主题的具体实现
        结合dwm层,来对访客主题进行轻量聚合.dws层 访客主题 要统计的指标 :  pv、uv、进入次数、跳出数、页面持续访问时长、
        轻量聚合实现思路 : 获取到数据源构建成流、按照我们要聚合的维度,进行分组、聚合实现的思路(开窗+具体的聚合操作reduce或者process)
        zk、kafka、logger、BaseLogApp、UniqueVisitorApp、UserJumpDetailApp、VisitorStatsApp
 */
public class VisitorStatsApp {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        // 设置检查点相关
         /*
        //2.1 开启检查点
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        //2.2 设置检查点超时时间
        env.getCheckpointConfig().setCheckpointTimeout(60000L);
        //2.3 设置取消job之后是否保留检查点
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        //2.4 设置重启策略
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,3000L));
        //2.5 设置状态后端
        env.setStateBackend(new FsStateBackend("xxxx"));
        //2.6 设置操作hadoop的用户
        System.setProperty("HADOOP_USER_NAME","atguigu");
        */

        //TODO 3.从kafka主题中读取数据
        //3.1 声明相关的主题以及消费者组
        String pageLogTopic = "dwd_page_log"; // pv、会话、页面持续时长
        String uvTopic = "dwm_unique_visitor"; // uv
        String ujdTopic = "dwm_user_jump_detail"; // 跳出用户
        String groupId = "visitor_stats_group"; //
        //3.2 创建消费者对象
        FlinkKafkaConsumer<String> pageLogKafkaSource = MyKafkaUtil.getKafkaSource(pageLogTopic, groupId);
        FlinkKafkaConsumer<String> uvKafkaSource = MyKafkaUtil.getKafkaSource(uvTopic, groupId);
        FlinkKafkaConsumer<String> ujdKafkaSource = MyKafkaUtil.getKafkaSource(ujdTopic, groupId);
        //3.3 消费数据  封装为流
        DataStreamSource<String> pageLogStrDS = env.addSource(pageLogKafkaSource);
        DataStreamSource<String> uvStrDS = env.addSource(uvKafkaSource);
        DataStreamSource<String> ujdStrDS = env.addSource(ujdKafkaSource);

//        pageLogStrDS.print(">>>");
//        uvStrDS.print("###");
//        ujdStrDS.print("$$$$");

        //TODO 4.对各个流的数据进行类型转换  jsonStr-->VisitorStats实体类对象
        /*
            这里将json字符串类型的数据,转化为 `VisitorStats` 对象格式,进行统一处理.
            这里我们要对`访客主题`中的不同指标进行聚合,再写入到olap数据库中,数据来源不同,需要我们构建多个流,并最终合并为一个流进行处理.
            多个流合并成一个流的方式 :
                1、使用 `connect算子`进行合并,<一国两制的方式合并>, 在处理过程中,流中处理的数据类型可以不同,并且只能对两个流进行和并.
                2、使用`union算子`进行合并,使用时需要注意:流中处理的数据格式必须相同,可以将多个流进行和并.
         */
        //4.1 pageLog
        SingleOutputStreamOperator<VisitorStats> pageLogStatsDS = pageLogStrDS.map(
                new MapFunction<String, VisitorStats>() {
                    @Override
                    public VisitorStats map(String jsonStr) throws Exception {
                        //将json字符串转换为json对象
                        JSONObject jsonObj = JSON.parseObject(jsonStr);
                        JSONObject commonJsonObj = jsonObj.getJSONObject("common");

                        JSONObject pageJsonObj = jsonObj.getJSONObject("page");
                        VisitorStats visitorStats = new VisitorStats(
                                "",
                                "",
                                commonJsonObj.getString("vc"), // 版本
                                commonJsonObj.getString("ch"),//渠道
                                commonJsonObj.getString("ar"),//地区
                                commonJsonObj.getString("is_new"),//是否新老用户标识
                                0L,
                                1L, // 没来一条数据,页面访问数量 加一
                                0L, //进入次数,这里指的是 进入系统的次数.如果访问的当前页面是首页,则进入次数 + 1
                                0L,
                                pageJsonObj.getLong("during_time"),
                                jsonObj.getLong("ts")
                        );

                        String lastPageId = pageJsonObj.getString("last_page_id");
                        if (lastPageId == null || lastPageId.length() == 0) {
                            visitorStats.setSv_ct(1L);
                        }
                        return visitorStats;
                    }
                }
        );
        pageLogStatsDS.print("==>page_loge<==");
        //4.2 uv
        SingleOutputStreamOperator<VisitorStats> uvStatsDS = uvStrDS.map(
                new MapFunction<String, VisitorStats>() {
                    @Override
                    public VisitorStats map(String jsonStr) throws Exception {
                        JSONObject jsonObj = JSON.parseObject(jsonStr);
                        JSONObject commonJsonObj = jsonObj.getJSONObject("common");
                        return new VisitorStats(
                                "",
                                "",
                                commonJsonObj.getString("vc"),
                                commonJsonObj.getString("ch"),
                                commonJsonObj.getString("ar"),
                                commonJsonObj.getString("is_new"),
                                1L, 0L, 0L, 0L, 0L, jsonObj.getLong("ts")
                        );
                    }
                }
        );
        uvStatsDS.print("==>uv<==");
        //4.3 ujd
        SingleOutputStreamOperator<VisitorStats> ujdStatDS = ujdStrDS.map(
                new MapFunction<String, VisitorStats>() {
                    @Override
                    public VisitorStats map(String jsonStr) throws Exception {
                        JSONObject jsonObj = JSON.parseObject(jsonStr);
                        JSONObject commonJsonObj = jsonObj.getJSONObject("common");
                        return new VisitorStats(
                                "",
                                "",
                                commonJsonObj.getString("vc"),
                                commonJsonObj.getString("ch"),
                                commonJsonObj.getString("ar"),
                                commonJsonObj.getString("is_new"),
                                0L, 0L, 0L, 1L, 0L, jsonObj.getLong("ts")
                        );
                    }
                }
        );
        ujdStatDS.print("==>user_just_detail<==");
        //TODO 5.将三条流的数据进行合并
        DataStream<VisitorStats> unionDS = pageLogStatsDS.union(
                uvStatsDS,
                ujdStatDS
        );
        unionDS.print("合并后的数据>>>>>");
        /*
            ==>page_loge<==:4>         VisitorStats(stt=, edt=, vc=v2.1.134, ch=xiaomi, ar=440000, is_new=1, uv_ct=0, pv_ct=1, sv_ct=0, uj_ct=0, dur_sum=11456, ts=1638870729000)
            ==>user_just_detail<==:1>  VisitorStats(stt=, edt=, vc=v2.1.134, ch=xiaomi, ar=110000, is_new=0, uv_ct=0, pv_ct=0, sv_ct=0, uj_ct=1, dur_sum=0, ts=1638870046000)
           合并后的数据>>>>>:4>          VisitorStats(stt=, edt=, vc=v2.1.134, ch=web, ar=110000, is_new=0, uv_ct=0, pv_ct=1, sv_ct=0, uj_ct=0, dur_sum=11804, ts=1638870728000)
         */
        //TODO 6.指定Watermark以及提取事件时间字段,设置水位线需要指定:单调递增or乱序程度.这里使用指定乱序程度.
        SingleOutputStreamOperator<VisitorStats> visitorStatsWithWatermarkDS = unionDS.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<VisitorStats>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<VisitorStats>() {
                                    @Override
                                    public long extractTimestamp(VisitorStats visitorStats, long recordTimestamp) {
                                        return visitorStats.getTs();
                                    }
                                }
                        )
        );

        //TODO 7.分组  注意：如果按照mid进行分组的话，单位时间内，不会有太多的操作，达不到聚合的效果。所以我们这里按照维度进行分组
        KeyedStream<VisitorStats, Tuple4<String, String, String, String>> keyedDS = visitorStatsWithWatermarkDS.keyBy(
                new KeySelector<VisitorStats, Tuple4<String, String, String, String>>() {
                    @Override
                    public Tuple4<String, String, String, String> getKey(VisitorStats visitorStats) throws Exception {
                        return Tuple4.of(
                                visitorStats.getVc(),
                                visitorStats.getAr(),
                                visitorStats.getCh(),
                                visitorStats.getIs_new()
                        );
                    }
                }
        );

        //TODO 8.开窗
        /*
            涉及到开窗操作,在Flink中提供了两种窗口概念 : 基于数量的窗口统计、基于时间(事件时间和处理时间)的窗口统计.
            并且窗口还会分为: 滑动窗口、滚动窗口、会话窗口.
            开窗后,我要对窗口本身的一些属性进行操作,又该怎么办呢 ?
            这里就需要在开窗后获取的流,再使用相关的算子.
                - 使用算子 `process算子`,从上下文对象中获取window对象,再获取具体属性.
                - 直接结合聚合操作的算子,比如 `reduce算子`,通过查看当前算子的构造器,可以看到,它允许我们传入`WindowFunction<>`这个函数对象,从而实现控制窗口相关的操作.
         */

        WindowedStream<VisitorStats, Tuple4<String, String, String, String>, TimeWindow> windowDS
                = keyedDS.window(TumblingEventTimeWindows.of(Time.seconds(30)));
        // 操作1、 窗口中的一些属性.可以直接使用通用的算子`process算子`.直接使用上下文对象获取到window对象,再进行操作
       /* SingleOutputStreamOperator<Object> process = windowDS.process(new ProcessWindowFunction<VisitorStats, Object, Tuple4<String, String, String, String>, TimeWindow>() {
            @Override
            public void process(Tuple4<String, String, String, String> stringStringStringStringTuple4, Context context, Iterable<VisitorStats> elements, Collector<Object> out) throws Exception {
                TimeWindow window = context.window();
                long start = window.getStart();
                long end = window.getEnd();
            }
        });*/

        //SingleOutputStreamOperator<Object> aggregate = windowDS.aggregate();

        //TODO 9.聚合计算
        /*
            开窗后的聚合操作,这里聚合的操作,有两种方式 :
            如果说,我窗口中当中两条进行聚合操作的数据格式一致,可以直接使用`reduce算子`.
            如果说,我窗口中两条进行聚合计算的数据格式不一致,需要自己指定聚合规则,则要使用更底层的算子`aggregate`,可以使用这个`aggregate算子`手动指定聚合过程.

         */
        SingleOutputStreamOperator<VisitorStats> reduceDS = windowDS.reduce(
                new ReduceFunction<VisitorStats>() {
                    @Override
                    public VisitorStats reduce(VisitorStats stats1, VisitorStats stats2) throws Exception {
                        // 注意这里的写法,不再创建一个新的对象,而是将一个`VisitorStats`作为当前计算的返回值,直接返回.但是在返回前,需要我们先对其中的属性进行更新操作,再返回.
                        stats1.setPv_ct(stats1.getPv_ct() + stats2.getPv_ct());
                        stats1.setUv_ct(stats1.getUv_ct() + stats2.getUv_ct());
                        stats1.setUj_ct(stats1.getUj_ct() + stats2.getUj_ct());
                        stats1.setDur_sum(stats1.getDur_sum() + stats2.getDur_sum());
                        stats1.setSv_ct(stats1.getSv_ct() + stats2.getSv_ct());
                        return stats1;
                    }
                },
                // 获取窗口的开始时间和结束时间.获取到的是时间戳格式的毫秒数.
                new WindowFunction<VisitorStats, VisitorStats, Tuple4<String, String, String, String>, TimeWindow>() {
                    @Override
                    public void apply(Tuple4<String, String, String, String> tuple4, TimeWindow window, Iterable<VisitorStats> elements, Collector<VisitorStats> out) throws Exception {
                        for (VisitorStats visitorStats : elements) {
                            visitorStats.setStt(DateTimeUtil.toYmdhms(new Date(window.getStart())));
                            visitorStats.setEdt(DateTimeUtil.toYmdhms(new Date(window.getEnd())));
                            visitorStats.setTs(System.currentTimeMillis());
                            out.collect(visitorStats);
                        }
                    }
                }
        );

        reduceDS.print(">>>>>");

        //TODO 10.将聚合的结果写到ClickHouse

        reduceDS.addSink(
                ClickHouseUtil.<VisitorStats>getSinkFunction("insert into visitor_stats_0609 values(?,?,?,?,?,?,?,?,?,?,?,?)")
        );

        env.execute();
    }
}
